package com.xbai.spark.core

import java.util

import org.apache.spark.rdd.RDD
import org.apache.spark.util.{AccumulatorV2, LongAccumulator}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 累加器：分布式共享只写数据
  * 累加器用来对信息进行聚合，通常在向 Spark 传递函数时，比如使用 map() 函数或者用 filter() 传条件时，
  * 可以使用驱动器程序中定义的变量，但是集群中运行的每个任务都会得到这些变量的一份新的副本，更新这些副本的值也不会影响驱动器中的对应变量。
  * 如果我们想实现所有分片处理时更新共享变量的功能，那么累加器可以实现我们想要的效果。
  * @author xbai
  * @Date 2021/1/4
  */
object Spark_Accumulator {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Accumulator")
    val sc = new SparkContext(conf)

    val dataRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
    // 创建累加器对象
    val accumulator: LongAccumulator = sc.longAccumulator
    dataRDD.foreach{
      case i => {
        // 执行累加器的累加功能
        accumulator.add(i)
      }
    }
    println("sum = " + accumulator.value)

    val rdd: RDD[String] = sc.makeRDD(List("hadoop", "hbase", "hive", "scala", "spark"), 2)
    // 创建累加器
    val wordAccumulator = new WordAccumulator
    // 注册累加器
    sc.register(wordAccumulator)
    rdd.foreach{
      case word => {
        wordAccumulator.add(word)
      }
    }
    println("res = " + wordAccumulator.value)
    sc.stop()
  }
}

/**
  * 自定义累加器
  * 1. 继承 AccumulatorV2
  * 2. 实现抽象方法
  * 3.创建累加器
  */
class WordAccumulator extends AccumulatorV2[String, util.ArrayList[String]] {

  private val list = new util.ArrayList[String]()

  // 当前累加器是否位初始化状态
  override def isZero: Boolean = list.isEmpty

  // 复制累加器对象
  override def copy(): AccumulatorV2[String, util.ArrayList[String]] = {
    new WordAccumulator
  }

  // 重置累加器对象
  override def reset(): Unit = list.clear()

  // 向累加器中增加数据
  override def add(v: String): Unit = {
    if (v.contains("h")) {
      list.add(v)
    }
  }

  // 合并累加器
  override def merge(other: AccumulatorV2[String, util.ArrayList[String]]): Unit = list.addAll(other.value)

  // 获取累加器的结果
  override def value: util.ArrayList[String] = list
}
